package engine

// 并发引擎结构体
type ConcurrentEngine struct {
	Scheduler        Scheduler // 任务调度器
	WorkerCount      int       // 并发数量
	ItemChan         chan Item //
	RequestProcessor Processor
}

type Processor func(Request) (ParseResult, error)

// 任务调度器结构体
type Scheduler interface {
	Submit(Request)           // 提交任务
	WorkerChan() chan Request // 请求任务
	Run()                     // 执行任务
	ReadyNotifier
}

type ReadyNotifier interface {
	WorkerReady(chan Request)
}

func (e *ConcurrentEngine) Run(seeds ...Request) {
	out := make(chan ParseResult) // worker的输出
	e.Scheduler.Run()

	// 创建 goruntine
	for i := 0; i < e.WorkerCount; i++ {
		// e.Scheduler.WorkerChan()：初始请求提交给 Scheduler
		e.createWorker(e.Scheduler.WorkerChan(), out, e.Scheduler)
	}

	// engine把请求任务提交给 Scheduler
	for _, r := range seeds {
		if isDuplicate(r.Url) {
			//fmt.Printf("Duplicate request: %s\n", r.Url)
			continue
		}
		e.Scheduler.Submit(r)
	}

	//itemCount := 0
	for {
		// 接收 Worker 输出结果
		result := <-out
		for _, item := range result.Items {
			//itemCount++
			//log.Printf("Got Item %d: %v", itemCount, item)
			//go func() { e.ItemChan <- item }()
			go func(i Item) {
				e.ItemChan <- i
			}(item)
		}

		// 把 Worker 解析出的 Request 再送给 Scheduler
		for _, request := range result.Requests {
			if isDuplicate(request.Url) {
				//fmt.Printf("Duplicate request: %s\n", request.Url)
				continue
			}
			e.Scheduler.Submit(request)
		}
	}
}

// 创建任务，调用worker，分发goroutine
func (e *ConcurrentEngine) createWorker(in chan Request, out chan ParseResult, ready ReadyNotifier) {
	go func() {
		for {
			ready.WorkerReady(in) // 告诉调度器任务空闲
			request := <-in
			//result, err := Worker(request)
			// 改成只调用服务: call rpc，不执行处理逻辑
			result, err := e.RequestProcessor(request)
			if err != nil {
				continue
			}

			// 将解析结果放入out
			out <- result
		}
	}()
}

// 存放已经获取的所有Url
var visitedUrls = make(map[string]bool)

// 判断Url是否重复
func isDuplicate(r string) bool {
	if visitedUrls[r] {
		return true
	}

	visitedUrls[r] = true
	return false
}
